package com.bckj.fastboot.mq.ons;

import com.alibaba.csp.sentinel.Entry;
import com.alibaba.csp.sentinel.SphU;
import com.alibaba.csp.sentinel.slots.block.BlockException;
import com.aliyun.openservices.ons.api.Action;
import com.aliyun.openservices.ons.api.Admin;
import com.aliyun.openservices.ons.api.Consumer;
import com.aliyun.openservices.ons.api.ExpressionType;
import com.aliyun.openservices.ons.api.MessageListener;
import com.aliyun.openservices.ons.api.MessageSelector;
import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.Producer;
import com.aliyun.openservices.ons.api.exception.ONSClientException;
import com.aliyun.openservices.ons.api.order.MessageOrderListener;
import com.aliyun.openservices.ons.api.order.OrderAction;
import com.aliyun.openservices.ons.api.order.OrderConsumer;
import com.aliyun.openservices.ons.api.order.OrderProducer;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.annotation.AnnotationUtils;

@Configuration
@ConditionalOnClass({OrderProducer.class, Producer.class})
@ConditionalOnProperty(
    prefix = "fastboot.ons.config",
    name = {"NAMESRV_ADDR", "GROUP_ID"}
)
@EnableConfigurationProperties({OnsProperties.class})
public class FastOnsAutoConfiguration {
    private final Map<String, Admin> consumerMap = new ConcurrentHashMap<>();

    public FastOnsAutoConfiguration() {
    }

    @Bean(
        initMethod = "start"
    )
    @ConditionalOnMissingBean({OrderProducer.class})
    public OrderProducer onsOrderProducer(OnsProperties properties) {
        return ONSFactory.createOrderProducer(properties.getConfig());
    }

    @Bean(
        initMethod = "start"
    )
    @ConditionalOnMissingBean({Producer.class})
    public Producer onsProducer(OnsProperties properties) {
        return ONSFactory.createProducer(properties.getConfig());
    }

    @Bean
    @ConditionalOnMissingBean({OrderConsumer.class})
    @ConditionalOnBean({MessageOrderListener.class})
    public List<OrderConsumer> onsOrderConsumer(OnsProperties properties, ApplicationContext context) {
        String group_id = properties.getConfig().getProperty("GROUP_ID");
        Map<String, MessageOrderListener> messageListenerMap = new HashMap<>();
        Map<String, Object> listenerBeanMap = context.getBeansWithAnnotation(OnsListener.class);
        listenerBeanMap.keySet().forEach((beanName) -> {
            Object bean = listenerBeanMap.get(beanName);
            Class<?> clazz = bean.getClass();
            OnsListener annotation = AnnotationUtils.findAnnotation(clazz, OnsListener.class);
            if (bean instanceof MessageOrderListener && annotation != null) {
                String gid = annotation.gid();
                if (gid.equals("")) {
                    gid = group_id;
                }

                OrderConsumer newConsumer;
                if (!this.consumerMap.containsKey(gid)) {
                    Properties config = new Properties();
                    config.putAll(properties.getConfig());
                    config.setProperty("GROUP_ID", gid);
                    newConsumer = ONSFactory.createOrderedConsumer(config);
                    newConsumer.start();
                    this.consumerMap.put(gid, newConsumer);
                } else {
                    Admin consumer = this.consumerMap.get(gid);
                    if (!(consumer instanceof OrderConsumer)) {
                        throw new ONSClientException("not order consumer. gid=" + gid);
                    }

                    newConsumer = (OrderConsumer)consumer;
                }

                String topic = annotation.topic();
                String subExpression = annotation.subExpression();
                ExpressionType type = annotation.type();
                MessageSelector messageSelector;
                if (type == ExpressionType.SQL92) {
                    messageSelector = MessageSelector.bySql(subExpression);
                } else {
                    messageSelector = MessageSelector.byTag(subExpression);
                }

                String key = String.format("topic:%s|sub:(%s)|type:%s", topic, subExpression, type.name());
                if (messageListenerMap.containsKey(gid + "|" + topic)) {
                    throw new ONSClientException("topic has exits. info=" + key);
                }

                newConsumer.subscribe(topic, messageSelector, this.handleMqOrderMessage(gid, topic, (MessageOrderListener)bean));
                messageListenerMap.put(gid + "|" + topic, (MessageOrderListener)bean);
            }

        });
        return this.consumerMap.values().stream().filter((consumer) -> consumer instanceof OrderConsumer).map((e) -> (OrderConsumer)e).collect(Collectors.toList());
    }

    @Bean
    @ConditionalOnMissingBean({Consumer.class})
    @ConditionalOnBean({MessageListener.class})
    public List<Consumer> onsConsumer(OnsProperties properties, ApplicationContext context) {
        String group_id = properties.getConfig().getProperty("GROUP_ID");
        Map<String, MessageListener> messageListenerMap = new HashMap<>();
        Map<String, Object> listenerBeanMap = context.getBeansWithAnnotation(OnsListener.class);
        listenerBeanMap.keySet().forEach((beanName) -> {
            Object bean = listenerBeanMap.get(beanName);
            Class<?> clazz = bean.getClass();
            OnsListener annotation = AnnotationUtils.findAnnotation(clazz, OnsListener.class);
            if (bean instanceof MessageListener && annotation != null) {
                String gid = annotation.gid();
                if (gid.equals("")) {
                    gid = group_id;
                }

                Consumer newConsumer;
                if (!this.consumerMap.containsKey(gid)) {
                    Properties config = new Properties();
                    config.putAll(properties.getConfig());
                    config.setProperty("GROUP_ID", gid);
                    newConsumer = ONSFactory.createConsumer(config);
                    newConsumer.start();
                    this.consumerMap.put(gid, newConsumer);
                } else {
                    Admin consumer = this.consumerMap.get(gid);
                    if (!(consumer instanceof Consumer)) {
                        throw new ONSClientException("not normal consumer. gid=" + gid);
                    }

                    newConsumer = (Consumer)consumer;
                }

                String topic = annotation.topic();
                String subExpression = annotation.subExpression();
                ExpressionType type = annotation.type();
                MessageSelector messageSelector;
                if (type == ExpressionType.SQL92) {
                    messageSelector = MessageSelector.bySql(subExpression);
                } else {
                    messageSelector = MessageSelector.byTag(subExpression);
                }

                String key = String.format("topic:%s|sub:(%s)|type:%s", topic, subExpression, type.name());
                if (messageListenerMap.containsKey(gid + "|" + topic)) {
                    throw new ONSClientException("topic has exits. info=" + key);
                }

                newConsumer.subscribe(topic, messageSelector, this.handleMqMessage(gid, topic, (MessageListener)bean));
                messageListenerMap.put(gid + "|" + topic, (MessageListener)bean);
            }

        });
        return this.consumerMap.values().stream().filter((consumer) -> consumer instanceof Consumer).map((e) -> (Consumer)e).collect(Collectors.toList());
    }

    MessageListener handleMqMessage(String groupId, String topic, MessageListener messageListener) {
        return (message, consumeContext) -> {
            try {
                Entry ignored = SphU.entry("MQ:NORMAL:" + groupId + ":" + topic);

                Action var6;
                try {
                    var6 = messageListener.consume(message, consumeContext);
                } catch (Throwable var9) {
                    if (ignored != null) {
                        try {
                            ignored.close();
                        } catch (Throwable var8) {
                            var9.addSuppressed(var8);
                        }
                    }

                    throw var9;
                }

                if (ignored != null) {
                    ignored.close();
                }

                return var6;
            } catch (BlockException var10) {
                System.err.println("Blocked, will retry later: " + message);
                return Action.ReconsumeLater;
            }
        };
    }

    MessageOrderListener handleMqOrderMessage(String groupId, String topic, MessageOrderListener messageOrderListener) {
        return (message, consumeOrderContext) -> {
            try {
                Entry ignored = SphU.entry("MQ:ORDER:" + groupId + ":" + topic);

                OrderAction var6;
                try {
                    var6 = messageOrderListener.consume(message, consumeOrderContext);
                } catch (Throwable var9) {
                    if (ignored != null) {
                        try {
                            ignored.close();
                        } catch (Throwable var8) {
                            var9.addSuppressed(var8);
                        }
                    }

                    throw var9;
                }

                if (ignored != null) {
                    ignored.close();
                }

                return var6;
            } catch (BlockException var10) {
                System.err.println("Blocked, will retry later: " + message);
                return OrderAction.Suspend;
            }
        };
    }
}
